package com.cold.datacollection.network;

import com.cold.common.domain.AlertMessage;
import com.cold.datacollection.config.KafkaConfig;
import com.cold.datacollection.util.SpringContextUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.text.SimpleDateFormat;

/**
 * 数据采集处理器
 */
public class DataCollectionHandler extends SimpleChannelInboundHandler<String> {

    private Logger logger = LoggerFactory.getLogger(DataCollectionHandler.class);

    private KafkaTemplate<String,String> kafkaTemplate = SpringContextUtil.getBean(KafkaTemplate.class);

    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        //解析报文 MSG|WHCC-A01|41|32|1|0.000000|0.000000|2024-03-01 10:22:56|V1.0
        try {
            logger.info("处理报文：" + s);
            String[] strs = s.split("\\|");
            AlertMessage message = new AlertMessage();
            message.setDeviceCode(strs[1]);
            message.setTem(Long.parseLong(strs[2]));
            message.setHum(Long.parseLong(strs[3]));
            message.setDeviceStatus(strs[4]);
            message.setLon(strs[5]);
            message.setLat(strs[6]);
            message.setSendTime(format.parse(strs[7]));
            //序列化为json
            String json = new ObjectMapper().writeValueAsString(message);
            //发送到kafka
            kafkaTemplate.send(KafkaConfig.TOPIC, json);
            logger.info("消息发送成功: " + json);
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }
}
